package net.bwie.zg6.spark

import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *京东用户日志离线分析
 *  todo: RDD 算子
 *     textFile
 *     map、distinct、reduceByKey、sortBy、sortByKey、foreach
 */
object SparkJdLogRddApp {

    def main(args: Array[String]): Unit = {
        // step1. 上下文环境
        val sc: SparkContext = SparkContext.getOrCreate(
            new SparkConf()
                .setAppName("SparkJdLogRddApp")
                .setMaster("local[1]")
        )
        sc.setLogLevel("WARN")

        // step2. 数据源-source todo 将要处理数据封装为RDD分布式数据集
        val logRdd: RDD[String] = sc.textFile(
            "zg6-spark-lecture/src/main/resources/jd-behavior.log",
            minPartitions = 2
        )
        logRdd.foreach(println)

        // step3. 数据转换-transformation  todo 调用RDD算子，处理数据
        // step4. 数据接收器-sink  todo 将结果RDD保存到外部存储系统


        // 4. 分析1：京东每日的UV，降序排序，打印输出；
        println("4. 分析1：京东每日的UV，降序排序，打印输出；")
        val rdd4: RDD[(String, Int)] = logRdd
            // s1.提取字段
            .map(line => {
                //字符串分割
                val split: Array[String] = line.split(",")
                // 日期
                val dateStr = split(3).substring(0, 10)
                // userId
                val userId = split(1)
                // 返回二元组
                (dateStr, userId)
            })
            // s2.去重
            .distinct()
            // s3.计数
            .map(tuple => (tuple._1, 1))
            .reduceByKey((tmp, item) => tmp + item)
            .sortBy(tuple => tuple._2, ascending = false)
        rdd4.foreach(println)

        // 5. 分析2：京东每日浏览量PV，降序排序，打印输出；
        println("5. 分析2：京东每日浏览量PV，降序排序，打印输出；")
        val rdd5: RDD[(Int, String)] = logRdd
            // s0. 过滤数据
            .filter(line => {
                //字符串分割
                val split: Array[String] = line.split(",")
                // 判断数据
                split.length == 4 & StringUtils.isNotEmpty(split(2))
            })
            // s1. 提取字段
            .map(line => {
                //字符串分割
                val split: Array[String] = line.split(",")
                // 日期
                val dateStr = split(3).substring(0, 10)
                // 返回二元组
                (dateStr, 1)
            })
            // s2.分组聚合
            .reduceByKey((tmp, item) => tmp + item)
            .map(tuple => tuple.swap)
            .sortByKey(ascending = false)
        rdd5.foreach(println)

        // step5. 关闭上下文
        sc.stop()
    }

}
